package com.wdciti.sccp.iot.springvertx.protocols.mqtt;

import com.wdciti.sccp.iot.springvertx.EndPointVerticle;
import io.vertx.core.Context;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.eventbus.DeliveryOptions;
import io.vertx.core.json.JsonArray;
import io.vertx.mqtt.MqttClient;
import io.vertx.mqtt.MqttClientOptions;
import lombok.extern.slf4j.Slf4j;

/**
 * @author JINWEN
 */
@Slf4j
public class SimpleMqttVerticle extends EndPointVerticle {

    private String host;
    private int port;
    private MqttClientOptions clientOptions = new MqttClientOptions();

    @Override
    public void init(Vertx vertx, Context context) {
        super.init(vertx, context);
        host = config().getString("host");
        port = config().getInteger("port");
        clientOptions = new MqttClientOptions(config().getJsonObject("options"));
    }

    @Override
    public void start() throws Exception {
        MqttClient mqttClient = MqttClient.create(vertx, clientOptions);
        mqttClient.connect(port, host).onComplete(result -> {
            if (result.succeeded()) {
                subscribeTopics(mqttClient);
                promiseReady();
            } else {
                result.cause().printStackTrace();
            }
        });

        vertx.eventBus().<Buffer>consumer(busAddr + EndPointVerticle.OUT).handler(hd -> {
            try {
                MqttHeader mqttHeader = MqttHeader.buildFrom(hd.headers());
                mqttClient.publish(mqttHeader.getTopicName(), hd.body(),
                                mqttHeader.getQosLevel(), mqttHeader.isDup(), mqttHeader.isRetain())
                        .onSuccess(success -> {
                            log.info("send to mqtt topic:{} success", mqttHeader.getTopicName());
                            log.debug("SimpleMqtt[Send]: opt:{}, body:{}", hd.headers().toString(), hd.body().toString());
                        })
                        .onFailure(fail -> fail.getCause().printStackTrace());
            } catch (RuntimeException e) {
                log.error("build mqtt header error!", e.getCause());
                e.printStackTrace();
            }
        });
    }

    private void subscribeTopics(MqttClient mqttClient) {
        JsonArray array = config().getJsonArray("topics");
        for (int i = 0; i < array.size(); i++) {
            MqttClientEndpointConfig.Topic topic = array.getJsonObject(i).mapTo(MqttClientEndpointConfig.Topic.class);
            mqttClient.subscribe(topic.getName(), topic.getQos())
                    .onComplete(ret -> log.info("subscribe:{},{}", topic, ret.succeeded()))
                    .onFailure(ret -> {
                        log.warn("subscribe ERROR!:{},{}", topic, ret.getCause().getMessage());
                        ret.printStackTrace();
                    });
        }
        mqttClient.publishHandler(hd -> {
            MqttHeader mqttHeader = MqttHeader.buildFrom(hd);
            DeliveryOptions opt = new DeliveryOptions();
            opt.setHeaders(mqttHeader.addMqttHeader(opt.getHeaders()));
            log.debug("SimpleMqtt[Receive], header:{}, body:{}", opt.toJson().toString(), hd.payload().toString());
            vertx.eventBus().send(busAddr + EndPointVerticle.IN, hd.payload(), opt);
        });
    }


    @Override
    public void stop() throws Exception {
        super.stop();
    }
}
